package com.atguigu.state;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by RUI on 2021/6/16 19:34
 */
public class State_Operator_List {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment
                .getExecutionEnvironment()
                .setParallelism(3);
        env.enableCheckpointing(3000);// 默认情况, flink不会自动checkpoint. 需要开启
        env
                .socketTextStream("hadoop162", 9999)
                .flatMap(new MyCountMapper())
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static class MyCountMapper implements FlatMapFunction<String, String>, CheckpointedFunction {

        List<String> list = new ArrayList<>();
        private ListState<String> listState;

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            for (String word : value.split(" ")) {
                list.add(word);
                out.collect(list.toString());
            }

            if (value.contains("a")) {
                throw new RuntimeException("手动抛异常");
            }
        }

        // Checkpoint时会调用这个方法，我们要实现具体的snapshot逻辑，比如将哪些本地状态持久化
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            System.out.println("snapshotState...");
            listState.clear();
            listState.update(list);
        }

        // 初始化时会调用这个方法，向本地状态中填充数据. 每个子任务调用一次
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("initializeState...");
            listState = context.getOperatorStateStore().getListState(new ListStateDescriptor<String>("listState", String.class));
            Iterable<String> strings = listState.get();
            for (String string : strings) {
                System.out.println("恢复:" + string);
                list.add(string);
            }
        }

    }
}
